// Copyright (c) 2022 Presto Labs Pte. Ltd.
// Author: daniel

#include "coin2/exchange/prex/order_v1/parser.h"

#include <memory>
#include <string>
#include <string_view>
#include <unordered_set>
#include <utility>
#include <vector>

#include <nlohmann/json.hpp>

#include "coin2/base/conversion.h"
#include "coin2/base/log.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::prex::order_v1 {

namespace {

using coin::proto::AccountBalance;
using coin::proto::CancelOrderResponse;
using coin::proto::ConversionBundle;
using coin::proto::ConversionElement;
using coin::proto::FillType;
using coin::proto::ProductFillBundle;
using coin::proto::ProductFillElement;
using coin::proto::ProductOrderBundle;
using coin::proto::ProductOrderElement;
using coin::proto::SubmitOrderResponse;
using coin2::exchange::base::order::IOrderContext;
using coin2::exchange::base::order::OrderSpec;
using coin2::exchange::prex::symbology::PrexCurrency;
using coin2::exchange::prex::symbology::PrexProduct;

using JsonValue = rapidjson::GenericValue<rapidjson::UTF8<>>;

inline int64_t GetProcOrderId(const JsonValue& data) {
  int64_t proc_order_id = 0;
  if (data.HasMember("clientOrderId")) {
    std::string proc_str = data["clientOrderId"].GetString();
    try {
      auto pos = proc_str.rfind("-");
      if (pos == std::string::npos) {
        proc_order_id = stoll(proc_str);
      } else {
        proc_order_id = stoll(proc_str.substr(pos + 1));
      }
    } catch (std::exception& e) {
      LOG(ERROR) << e.what() << ": " << JsonToString(data);
    }
  }
  return proc_order_id;
}

inline coin::proto::OrderSide GetOrderSide(const std::string_view& side) {
  if (side == "ORDER_SIDE_BUY") {
    return coin::proto::OrderSide::BUY_ORDER;
  } else if (side == "ORDER_SIDE_SELL") {
    return coin::proto::OrderSide::SELL_ORDER;
  } else {
    return coin::proto::OrderSide::UNKNOWN_ORDER_SIDE;
  }
}

inline coin::proto::OrderType GetOrderType(const std::string& type) {
  if (type == "ORDER_TYPE_LIMIT") {
    return coin::proto::OrderType::LIMIT_ORDER;
  } else if (type == "ORDER_TYPE_MARKET") {
    return coin::proto::OrderType::MARKET_ORDER;
  } else if (type == "ORDER_TYPE_LIQUIDATION") {
    return coin::proto::OrderType::LIQUIDATION_ORDER;
  } else if (type == "ORDER_TYPE_TAKEOVER") {
    return coin::proto::OrderType::BLP_TAKEOVER_ORDER;
  } else {
    return coin::proto::OrderType::UNKNOWN_ORDER_TYPE;
  }
}

inline void SetOrderStatus(const std::string_view& status, ProductOrderElement* order) {
  if (status == "ORDER_STATUS_NEW" || status == "ORDER_STATUS_PARTIALLY_FILLED") {
    order->set_is_live(true);
  } else if (status == "ORDER_STATUS_REJECTED") {
    order->set_cancel_confirmed(true);  // submit no reject response (temporary handle)
    // order->set_is_live(false);
  } else if (status == "ORDER_STATUS_CANCELED") {
    order->set_cancel_confirmed(true);
  } else if (status == "ORDER_STATUS_FILLED") {
    order->set_fully_filled(true);
  } else {
    LOG(ERROR) << "[ParseOrderInfo] Unknown order status! " << status;
  }
}

inline std::string GetPrexExchangeOrderErrorMsg(
    const rapidjson::GenericValue<rapidjson::UTF8<>>& json) {
  if (json.HasMember("error")) {
    return json["error"].GetString();
  } else {
    return {};
  }
}

inline coin::proto::OrderErrorCode GetOrderErrorCodeFromPrex(
    const rapidjson::GenericValue<rapidjson::UTF8<>>& json_error_msg) {
  using coin::proto::OrderErrorCode;
  auto result = coin::proto::OrderErrorCode::ORDER_UNKNOWN_ERROR;
  std::string_view error_code = json_error_msg.GetString();

  if (error_code == "0") {
    result = OrderErrorCode::ORDER_NO_ERROR;
  } else if (error_code == "Account has no [XBt]") {
    result = OrderErrorCode::SUBMIT_ORDER_INSUFFICIENT_FUND;
  } else if (error_code == "Account has insufficient balance for requested action.") {
    result = OrderErrorCode::SUBMIT_ORDER_INSUFFICIENT_FUND;
  } else if (error_code == "Invalid ordStatus") {
    result = OrderErrorCode::ORDER_NOT_FOUND;
  } else if (error_code == "Not Found") {
    result = OrderErrorCode::ORDER_NOT_FOUND;
  }

  return result;
}

bool IsPrexErrMessage(const JsonValue& doc) { return doc.HasMember("error"); }

bool IsWallet(const rapidjson::GenericValue<rapidjson::UTF8<>>& json) {
  if (!json.HasMember("topic") || !json.HasMember("data")) {
    return false;
  }
  const auto topic = std::string_view(json["topic"].GetString());
  return topic == "private/wallet";
}

bool IsOrderUpdate(const rapidjson::GenericValue<rapidjson::UTF8<>>& json) {
  if (!json.HasMember("topic") || !json.HasMember("data")) {
    return false;
  }
  const auto topic = std::string_view(json["topic"].GetString());
  return topic == "private/order";
}

bool IsExecutionUpdate(const rapidjson::GenericValue<rapidjson::UTF8<>>& json) {
  if (!json.HasMember("topic") || !json.HasMember("data")) {
    return false;
  }
  const auto topic = std::string_view(json["topic"].GetString());
  return topic == "private/execution";
}

}  // namespace

bool PrexParser::LoadZombieOrderCheckList(const std::string& mixin_order_ids) {
  if (mixin_order_ids.empty()) {
    return false;
  }

  mixin_zombie_order_ids_map_.clear();
  std::vector<std::string> vec;
  boost::split(vec, mixin_order_ids, boost::is_any_of(","));
  for (const auto& ele : vec) {
    if (ele.find("@") == std::string::npos) {
      continue;
    }
    std::vector<std::string> ids;
    boost::split(ids, ele, boost::is_any_of("@"));
    if (ids[0].empty() || ids[1].empty()) {
      continue;
    }
    const auto proc_order_id = stoll(ids[0]);
    const auto& exchange_order_id = ids[1];
    mixin_zombie_order_ids_map_[exchange_order_id].push_back(proc_order_id);
  }

  return mixin_zombie_order_ids_map_.size() > 0;
}

std::string PrexParser::ParseAuthToken(const JsonValue& doc) {
  if (doc.IsObject() && doc.HasMember("token")) {
    return doc["token"].GetString();
  }
  return "";
}

bool PrexParser::ParseRawBalance(const JsonValue& doc, RawBalanceMap* raw_balance_map) {
  /*
  {
    "wallet": [
      {
        "accountId": "100106",
        "currency": "USD",
        "walletBalance": "1000000000.002071824",
        "walletBalanceInUsd": "1000000000.002071824",
        "walletBalanceAvailable": "1000000000.002071824",
        "walletBalanceAvailableInUsd": "1000000000.002071824",
        "walletBalanceExclWithdrawalPending": "1000000000.002071824",
        "walletBalanceExclWithdrawalPendingInUsd": "1000000000.002071824",
        "walletBalanceExclLocked": "999999998.012073814",
        "walletBalanceExclLockedInUsd": "999999998.012073814",
        "walletBalanceWithdrawable": "999999998.012073814",
        "walletBalanceWithdrawableInUsd": "999999998.012073814",
        "withdrawalPending": "0",
        "withdrawalPendingInUsd": "0",
        "netRealizedPnl": "0.002071824"
      },
      ...
    ]
  }
  */
  if (!doc.IsObject() || IsPrexErrMessage(doc)) {
    LOG(ERROR) << "[ParseRawBalance] " << JsonToString(doc);
    return false;
  }
  if (!doc.HasMember("wallet") || !doc["wallet"].IsArray()) {
    LOG(ERROR) << "[ParseRawBalance] " << JsonToString(doc);
    return false;
  }

  for (const auto& data : doc["wallet"].GetArray()) {
    if (!data.HasMember("currency") || !data.HasMember("walletBalance") ||
        !data.HasMember("walletBalanceExclLocked")) {
      continue;
    }
    const std::string native_currency = data["currency"].GetString();
    if (balance_cache_map_.count(native_currency) == 0) {
      CurrencyBalance balance;
      balance.set_currency_native(native_currency);
      balance.set_total(0);
      balance.set_hold(0);
      balance.set_available(0);
      balance_cache_map_[native_currency] = balance;
    }
    auto& balance = balance_cache_map_[native_currency];
    auto total = stod(data["walletBalance"].GetString());
    auto avail = stod(data["walletBalanceExclLocked"].GetString());
    balance.set_total(total);
    balance.set_hold(total - avail);
    balance.set_available(avail);
    raw_balance_map->emplace(balance.currency_native(), balance);
  }

  return true;
}

// NOTE: when field exchange and market_type are set, which means parsed successfully
auto PrexParser::ParseBalance(const JsonValue& doc) -> AccountBalance {
  RawBalanceMap raw_balance_map;
  auto success = ParseRawBalance(doc, &raw_balance_map);
  if (!success) {
    return {};
  }

  AccountBalance acc_balance;
  acc_balance.set_exchange("Prex");
  acc_balance.set_market_type("Spot");
  for (const auto& native_currency : symcache_->GetCurrenciesNativeUpper()) {
    auto* balance = acc_balance.add_each_balance();
    if (raw_balance_map.count(native_currency) == 0) {
      balance->set_total(0);
      balance->set_hold(0);
      balance->set_available(0);
    } else {
      balance->MergeFrom(raw_balance_map.at(native_currency));
    }
    auto* currency = symcache_->GetCurrencyByNative(native_currency);
    balance->set_currency(currency->currency());
    balance->set_currency_native(currency->native_currency());
    DLOG(INFO) << "BALANCE: " << currency->native_currency() << " " << balance->total() << " "
               << balance->available();
  }
  is_account_balance_initialized_ = true;

  return acc_balance;
}

void PrexParser::ParseRestAccountBalance() {
  JsonDocument doc;
  if (!PrepareJsonDocument(&doc)) {
    LOG(ERROR) << "[ParseRestAccountBalance] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  auto account_balance = ParseBalance(doc);
  if (!account_balance.has_exchange() || !account_balance.has_market_type()) {
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  result_.set_type(MsgType::ACCOUNT_BALANCE);
  *result_.mutable_account_balance() = std::move(account_balance);
}

bool PrexParser::ParseWsRawBalance(const JsonValue& doc, RawBalanceMap* raw_balance_map) {
  /*
  {
    "topic": "private/wallet",
    "data": [
      {
        "actionType": "SNAPSHOT",
        "rows": [
          {
            "accountId": "100106",
            "currency": "USD",
            "walletBalance": "1000000000.002071824",
            "walletBalanceInUsd": "1000000000.002071824",
            "walletBalanceAvailable": "1000000000.002071824",
            "walletBalanceAvailableInUsd": "1000000000.002071824",
            "walletBalanceExclWithdrawalPending": "1000000000.002071824",
            "walletBalanceExclWithdrawalPendingInUsd": "1000000000.002071824",
            "walletBalanceExclLocked": "999999998.012073814",
            "walletBalanceExclLockedInUsd": "999999998.012073814",
            "walletBalanceWithdrawable": "999999998.012073814",
            "walletBalanceWithdrawableInUsd": "999999998.012073814",
            "withdrawalPending": "0",
            "withdrawalPendingInUsd": "0",
            "netRealizedPnl": "0.002071824"
          },
          ...
        ]
      }
    ]
  }
  {
    "topic": "private/wallet",
    "data": [
      {
        "actionType": "UPDATE",
        "rows": [
          {
            "walletBalanceExclLocked": "999999998.013435174",
            "walletBalanceExclLockedInUsd": "999999998.013435174",
            "walletBalanceWithdrawable": "999999998.013435174",
            "walletBalanceWithdrawableInUsd": "999999998.013435174",
            "currency": "USD"
          }
        ]
      }
    ]
  }
  */
  if (!doc.HasMember("data") || !doc["data"].IsArray()) {
    LOG(ERROR) << JsonToString(doc);
    return false;
  }

  for (const auto& info : doc["data"].GetArray()) {
    std::string_view action = info["actionType"].GetString();
    if (action == "DELETE") {
      continue;
    }
    if (action == "SNAPSHOT") {
      for (const auto& row : info["rows"].GetArray()) {
        if (!row.HasMember("walletBalance") || !row.HasMember("walletBalanceExclLocked")) {
          continue;
        }
        const std::string native_currency = row["currency"].GetString();
        if (balance_cache_map_.count(native_currency) == 0) {
          CurrencyBalance balance;
          balance.set_currency_native(native_currency);
          balance.set_total(0);
          balance.set_hold(0);
          balance.set_available(0);
          balance_cache_map_[native_currency] = std::move(balance);
        }
        auto& balance = balance_cache_map_[native_currency];
        auto total = stod(row["walletBalance"].GetString());
        auto avail = stod(row["walletBalanceExclLocked"].GetString());
        balance.set_total(total);
        balance.set_hold(total - avail);
        balance.set_available(avail);
        raw_balance_map->emplace(balance.currency_native(), balance);
      }
      is_account_balance_initialized_ = true;
    } else {  // UPDATE
      for (const auto& row : info["rows"].GetArray()) {
        if (!row.HasMember("walletBalanceExclLocked") && !row.HasMember("walletBalance")) {
          continue;
        }
        const std::string native_currency = row["currency"].GetString();
        if (balance_cache_map_.count(native_currency) == 0) {
          continue;
        }

        auto& balance = balance_cache_map_[native_currency];
        auto total = balance.total();
        if (row.HasMember("walletBalance")) {
          total = stod(row["walletBalance"].GetString());
        }
        auto avail = balance.available();
        if (row.HasMember("walletBalanceExclLocked")) {
          avail = stod(row["walletBalanceExclLocked"].GetString());
        }
        balance.set_total(total);
        balance.set_hold(total - avail);
        balance.set_available(avail);
        raw_balance_map->emplace(balance.currency_native(), balance);
      }
    }
  }

  return true;
}

AccountBalance PrexParser::ParseWsBalance(const JsonValue& doc) {
  RawBalanceMap raw_balance_map;
  bool success = ParseWsRawBalance(doc, &raw_balance_map);
  if (!success) {
    return {};
  }
  // update account balance, just publish delta change from ws update
  AccountBalance acc_balance;
  acc_balance.set_exchange("Prex");
  acc_balance.set_market_type("Spot");
  for (const auto& native_currency : symcache_->GetCurrenciesNativeUpper()) {
    if (raw_balance_map.count(native_currency) == 0) {
      continue;
    }
    auto* balance = acc_balance.add_each_balance();
    balance->MergeFrom(raw_balance_map.at(native_currency));
    auto* currency = symcache_->GetCurrencyByNative(native_currency);
    balance->set_currency(currency->currency());
    balance->set_currency_native(currency->native_currency());
    DLOG(INFO) << "BALANCE: " << currency->native_currency() << " " << balance->total() << " "
               << balance->available();
  }

  return acc_balance;
}

std::tuple<bool, ProductOrderElement> PrexParser::ParseOrderInfo(const JsonValue& data) {
  /*
  {
    "hasNext": false,
    "history": [
      {
        "orderId": "3a6e2ec1-19a5-1c16-3a6e-2ec119a51c16",
        "symbol": "BTCUSD.PERP",
        "clientOrderId": "",
        "side": "ORDER_SIDE_BUY",
        "orderType": "ORDER_TYPE_LIMIT",
        "price": "6000.500000000000000000",
        "orderQty": "0.100000000000000000",
        "timeInForce": "TIME_IN_FORCE_IMMEDIATE_OR_CANCEL",
        "execInst": "EXEC_INST_PARTICIPATE_DO_NOT_INITIATE",
        "specialFlag": "SPECIAL_FLAG_INVALID",
        "text": "Web",
        "orderTime": 1593329898217934800,
        "transactTime": 1593329898217934800,
        "orderStatus": "ORDER_STATUS_CANCELED",
        "leavesQty": "0.100000000000000000",
        "cumQty": "0.000000000000000000",
        "avgPrice": "0.000000",
        "cumFee": "0.000000000000000000"
      }
    ]
  }
  */
  ProductOrderElement order;
  std::string exchange_order_id = data["orderId"].GetString();

  bool normal =
      (std::string_view(data["orderType"].GetString()) != "ORDER_TYPE_UNKNOWN_NOT_ACTIVE");

  if (normal) {
    order.set_market_type(MarketType::Spot);
    order.set_exchange_type(ExchangeType::Prex);
    order.set_api_version("v1");
    order.set_exchange_order_id(exchange_order_id);
    order.set_symbol(data["symbol"].GetString());
    order.set_price(stod(data["price"].GetString()));
    order.set_qty(stod(data["orderQty"].GetString()));
    order.set_proc_order_id(GetProcOrderId(data));
    order.set_side(GetOrderSide(data["side"].GetString()));
    order.set_order_type(GetOrderType(data["orderType"].GetString()));
    order.set_timestamp(msg_->timestamp);
    SetOrderStatus(data["orderStatus"].GetString(), &order);
    return {true, order};
  }

  auto* oc = ocm_->FindByExchangeOrderId(exchange_order_id);
  if (oc) {
    const OrderSpec& spec = oc->order_spec();
    order.set_market_type(MarketType::Spot);
    order.set_exchange_type(ExchangeType::Prex);
    order.set_api_version("v1");
    order.set_exchange_order_id(exchange_order_id);
    order.set_symbol(spec.product);
    order.set_price(spec.price);
    order.set_qty(spec.qty);
    order.set_proc_order_id(oc->proc_order_id());
    order.set_side(spec.order_side);
    order.set_order_type(spec.order_type);
    order.set_cancel_confirmed(true);  // (TODO, check carefully).
    return {true, order};
  }

  LOG(ERROR) << "[ParseOrderInfo] Unknown orders were cancelled" << JsonToString(data);
  return {false, {}};
}

void PrexParser::ParseRestOpenOrders() {
  /*
  {
    "order": [
      {
        "symbol": "BTCUSD.PERP",
        "orderId": "477d2f54-1f05-a873-f7ce-7b0934dbcb84",
        "clientOrderId": "1625024939669862949",
        "side": "ORDER_SIDE_BUY",
        "orderType": "ORDER_TYPE_LIMIT",
        "price": "35092.4",
        "orderQty": "0.001",
        "timeInForce": "TIME_IN_FORCE_GOOD_TILL_CANCEL",
        "execInst": "EXEC_INST_PARTICIPATE_DO_NOT_INITIATE",
        "orderTime": "1625024983597981332",
        "transactTime": "1625024983597981332",
        "orderStatus": "ORDER_STATUS_NEW",
        "leavesQty": "0.001",
        "cumQty": "0",
        "avgPrice": "0",
        "cumFee": "0",
        "specialFlag": "SPECIAL_FLAG_INVALID",
        "text": "",
        "orderCancelReason": "ORDER_CANCEL_REASON_INVALID"
      },
      {
        "symbol": "BTCUSD.PERP",
        "orderId": "a1ade989-5c9e-67af-608e-06ef10b0de86",
        "clientOrderId": "1625024939669862948",
        "side": "ORDER_SIDE_SELL",
        "orderType": "ORDER_TYPE_LIMIT",
        "price": "35444.9",
        "orderQty": "0.001",
        "timeInForce": "TIME_IN_FORCE_GOOD_TILL_CANCEL",
        "execInst": "EXEC_INST_PARTICIPATE_DO_NOT_INITIATE",
        "orderTime": "1625024983488750489",
        "transactTime": "1625024983488750489",
        "orderStatus": "ORDER_STATUS_NEW",
        "leavesQty": "0.001",
        "cumQty": "0",
        "avgPrice": "0",
        "cumFee": "0",
        "specialFlag": "SPECIAL_FLAG_INVALID",
        "text": "",
        "orderCancelReason": "ORDER_CANCEL_REASON_INVALID"
      }
    ]
  }
  */
  JsonDocument data;
  if (!PrepareJsonDocument(&data)) {
    LOG(ERROR) << "[ParseRestOpenOrders] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  if (data.IsObject() && IsPrexErrMessage(data)) {
    LOG(ERROR) << "[ParseRestOpenOrders] " << JsonToString(data);
    return;
  }
  if (!data.HasMember("order") || !data["order"].IsArray()) {
    LOG(ERROR) << "[ParseRestOpenOrders] Unexpected order list format!" << JsonToString(data);
    return;
  }

  result_.set_type(MsgType::ORDER_STATUS);
  auto product_order_bundle = result_.mutable_order_list();
  product_order_bundle->Clear();
  for (const auto& info : data["order"].GetArray()) {
    auto [success, new_order] = ParseOrderInfo(info);
    if (success) {
      auto* order = product_order_bundle->add_each_order();
      order->MergeFrom(new_order);
    }
  }

  CheckZombieOrder(product_order_bundle);
}

void PrexParser::ParseRestAllOrders() {
  JsonDocument data;
  if (!PrepareJsonDocument(&data)) {
    LOG(ERROR) << "[ParseRestAllOrders] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  if (data.IsObject() && IsPrexErrMessage(data)) {
    LOG(ERROR) << "[ParseRestAllOrders] " << JsonToString(data);
    return;
  }

  if (!data.HasMember("history") || !data["history"].IsArray()) {
    LOG(ERROR) << "[ParseRestAllOrders] Unexpected order list format!" << JsonToString(data);
    return;
  }

  LoadZombieOrderCheckList(msg_->http_context->extra_info);

  result_.set_type(MsgType::ORDER_STATUS);
  auto product_order_bundle = result_.mutable_order_list();
  product_order_bundle->Clear();
  for (const auto& info : data["history"].GetArray()) {
    auto [success, new_order] = ParseOrderInfo(info);
    if (success) {
      auto* order = product_order_bundle->add_each_order();
      order->MergeFrom(new_order);
    }
  }

  CheckZombieOrder(product_order_bundle);
}

void PrexParser::CheckZombieOrder(ProductOrderBundle* product_order_bundle) {
  if (product_order_bundle->each_order_size() == 0) {
    return;
  }
  if (mixin_zombie_order_ids_map_.size() == 0) {
    return;
  }

  std::vector<std::pair<std::string, int64_t>> mixin_order_ids;
  for (const auto& order_info : product_order_bundle->each_order()) {
    mixin_order_ids.push_back(
        std::make_pair(order_info.exchange_order_id(), order_info.proc_order_id()));
  }
  // set dead for the order which proc_order_id is difference, which means the order is amended
  for (const auto& [exchange_order_id, client_order_id] : mixin_order_ids) {
    if (mixin_zombie_order_ids_map_.count(exchange_order_id) == 0) {
      continue;
    }
    auto& proc_order_ids = mixin_zombie_order_ids_map_.at(exchange_order_id);
    auto iter = proc_order_ids.begin();
    while (iter != proc_order_ids.end()) {
      auto proc_order_id = *iter;
      iter = proc_order_ids.erase(iter);
      if (proc_order_id == client_order_id) {
        continue;
      }
      auto* oc = ocm_->FindByProcOrderId(proc_order_id);
      if (!oc) {
        continue;
      }
      if (!ocm_->IsInWorkingOrders(oc)) {
        continue;
      }
      LOG(INFO) << "set zombie order dead " << proc_order_id << "@" << oc->exchange_order_id();
      auto* order = product_order_bundle->add_each_order();
      order->set_market_type(MarketType::Spot);
      order->set_exchange_type(ExchangeType::Prex);
      order->set_api_version("v1");
      order->set_exchange_order_id(oc->exchange_order_id());
      order->set_symbol(oc->order_spec().product);
      order->set_price(oc->order_spec().price);
      order->set_qty(oc->order_spec().qty);
      order->set_proc_order_id(oc->proc_order_id());
      order->set_side(oc->order_spec().order_side);
      order->set_order_type(oc->order_spec().order_type);
      order->set_timestamp(msg_->timestamp);
      order->set_cancel_confirmed(true);
    }
    mixin_zombie_order_ids_map_.erase(exchange_order_id);
  }
}

// parse order info from "private/order" ws message
ProductOrderBundle PrexParser::ParseWsOrderInfoListV2(const JsonValue& data) {
  /*
  {
    "topic": "private/order",
    "data": [
      {
        "actionType": "SNAPSHOT",
        "rows": [
          {
            "accountId": "9800101",
            "symbol": "BTCUSD.PERP",
            "orderId": "f3c18c2e-6fa1-1c16-f3c1-8c2e6fa11c16",
            "side": "ORDER_SIDE_BUY",
            "orderType": "ORDER_TYPE_LIMIT",
            "price": "6000.5",
            "orderQty": "0.1",
            "timeInForce": "TIME_IN_FORCE_GOOD_TILL_CANCEL",
            "execInst": "EXEC_INST_PARTICIPATE_DO_NOT_INITIATE",
            "text": "Web",
            "orderTime": 1593325867078948685,
            "transactTime": 1593325867078948685,
            "orderStatus": "ORDER_STATUS_NEW",
            "leavesQty": "0.1",
            "cumQty": "0",
            "avgPrice": "0",
            "cumFee": "0",
            "clientOrderId": "",
            "specialFlag": "SPECIAL_FLAG_INVALID"
          }
        ]
      }
    ]
  }
  {
    "topic":"private/order",
    "data":[
        {
            "actionType":"INSERT",
            "rows":[
                {
                    "accountId":"9800101",
                    "symbol":"BTCUSD.PERP",
                    "orderId":"36c9cb16-71ec-1c16-36c9-cb1671ec1c16",
                    "clientOrderId":"asdf1234",
                    "side":"ORDER_SIDE_BUY",
                    "orderType":"ORDER_TYPE_LIMIT",
                    "price":"6100.5",
                    "orderQty":"0.1",
                    "timeInForce":"TIME_IN_FORCE_GOOD_TILL_CANCEL",
                    "execInst":"EXEC_INST_PARTICIPATE_DO_NOT_INITIATE",
                    "text":"Web",
                    "orderTime":"1593408338642650583",
                    "transactTime":"1593408338642650583",
                    "orderStatus":"ORDER_STATUS_NEW",
                    "leavesQty":"0.1",
                    "cumQty":"0",
                    "avgPrice":"0",
                    "cumFee":"0",
                    "specialFlag":"SPECIAL_FLAG_INVALID"
                }]
        }]
  }
  {
    "topic":"private/order",
    "data":[
        {
            "actionType":"UPDATE",
            "rows":[
                {
                    "transactTime":"1593409302124825175",
                    "orderStatus":"ORDER_STATUS_CANCELED",
                    "orderId":"064a46c8-41ed-1c16-064a-46c841ed1c16"
                }]
        }]
  }
  {
    "topic":"private/order",
    "data":[
        {
            "actionType":"DELETE",
            "rows":[
                {
                    "orderId":"36c9cb16-71ec-1c16-36c9-cb1671ec1c16"
                }]
        }]
  }
  */
  ProductOrderBundle product_order_bundle;

  if (!(data.HasMember("data") && data["data"].IsArray())) {
    LOG(ERROR) << "Unexpected ws order list format!" << JsonToString(data);
    return product_order_bundle;
  }

  for (const auto& info : data["data"].GetArray()) {
    for (const auto& row : info["rows"].GetArray()) {
      if (row.HasMember("orderStatus")) {
        auto* order = product_order_bundle.add_each_order();
        order->set_market_type(MarketType::Spot);
        order->set_exchange_type(ExchangeType::Prex);
        order->set_api_version("v1");
        std::string exchange_id = row["orderId"].GetString();
        if (row.HasMember("symbol")) {
          order->set_symbol(row["symbol"].GetString());
        } else {
          auto* oc = GetOcm()->FindByExchangeOrderId(exchange_id);
          if (oc != nullptr) {
            order->set_symbol(oc->order_spec().product);
            DLOG(INFO) << "no symbol, set oc symbol:" << oc->order_spec().product;
          }
        }
        if (row.HasMember("price")) {
          order->set_price(stod(row["price"].GetString()));
        }
        if (row.HasMember("orderQty")) {
          order->set_qty(stod(row["orderQty"].GetString()));
        }
        if (row.HasMember("side")) {
          order->set_side(GetOrderSide(row["side"].GetString()));
        }
        if (row.HasMember("orderType")) {
          order->set_order_type(GetOrderType(row["orderType"].GetString()));
        }
        order->set_exchange_order_id(exchange_id);
        order->set_proc_order_id(GetProcOrderId(row));
        if (row.HasMember("transactTime")) {
          order->set_timestamp(stoll(row["transactTime"].GetString()));
        } else {
          order->set_timestamp(GetCurrentTimestamp());
        }
        SetOrderStatus(row["orderStatus"].GetString(), order);
      }
    }
  }
  return product_order_bundle;
}

// parse order info from "private/execution" ws message
ProductOrderBundle PrexParser::ParseWsOrderInfoListV3(const JsonValue& data) {
  /*
  "private_exchange_message": {
    "type": "WS_MESSAGE",
    "raw_msg_id": "936982463203247223",
    "receive_timestamp": "1622427877793041122",
    "raw_bytes": {
      "topic": "private/execution",
      "data": {
        "actionType": "INSERT",
        "rows": [
          {
            "execId": "66308d23-cd48-5ce1-f630-070e340f440c",
            "execType": "EXEC_TYPE_NEW",
            "orderId": "8fb527c1-064c-73cc-0e11-8406eff8d9c5",
            "clientOrderId": "1622427654544879177",
            "accountId": "101124",
            "symbol": "BTCUSD.PERP",
            "side": "ORDER_SIDE_SELL",
            "orderType": "ORDER_TYPE_LIMIT",
            "price": "34736.3",
            "orderQty": "0.001",
            "timeInForce": "TIME_IN_FORCE_GOOD_TILL_CANCEL",
            "execInst": "EXEC_INST_PARTICIPATE_DO_NOT_INITIATE",
            "settlementCurrency": "USD",
            "orderTime": 1622427877482433500,
            "transactTime": 1622427877482433500,
            "orderStatus": "ORDER_STATUS_NEW",
            "leavesQty": "0.001",
            "cumQty": "0",
            "avgPrice": "0",
            "lastPrice": "0",
            "lastQty": "0",
            "execFee": "0",
            "feeRate": "0",
            "specialFlag": "SPECIAL_FLAG_INVALID",
            "text": "",
            "orderCancelReason": "ORDER_CANCEL_REASON_INVALID",
            "matchId": "",
            "matchTime": 0,
            "aggressorIndicator": false
          }
        ]
      }
    }
  }
  "private_exchange_message": {
    "type": "WS_MESSAGE",
    "raw_msg_id": "936982463203247242",
    "receive_timestamp": "1622427881763449704",
    "raw_bytes": {
      "topic": "private/execution",
      "data": {
        "actionType": "INSERT",
        "rows": [
          {
            "execId": "5681f91e-4a59-4d13-44d8-5712e2410656",
            "execType": "EXEC_TYPE_CANCELED",
            "orderId": "8fb527c1-064c-73cc-0e11-8406eff8d9c5",
            "clientOrderId": "1622427654544879177",
            "accountId": "101124",
            "symbol": "BTCUSD.PERP",
            "side": "ORDER_SIDE_SELL",
            "orderType": "ORDER_TYPE_LIMIT",
            "price": "34736.3",
            "orderQty": "0.001",
            "timeInForce": "TIME_IN_FORCE_GOOD_TILL_CANCEL",
            "execInst": "EXEC_INST_PARTICIPATE_DO_NOT_INITIATE",
            "settlementCurrency": "USD",
            "orderTime": 1622427877482433500,
            "transactTime": 1622427881252119600,
            "orderStatus": "ORDER_STATUS_CANCELED",
            "leavesQty": "0.001",
            "cumQty": "0",
            "avgPrice": "0",
            "lastPrice": "0",
            "lastQty": "0",
            "execFee": "0",
            "feeRate": "0",
            "orderCancelReason": "ORDER_CANCEL_REASON_USER_REQUESTED",
            "specialFlag": "SPECIAL_FLAG_INVALID",
            "text": "",
            "matchId": "",
            "matchTime": 0,
            "aggressorIndicator": false
          }
        ]
      }
    }
  }
  */
  ProductOrderBundle product_order_bundle;

  if (!(data.HasMember("data") && data["data"].IsObject())) {
    LOG(ERROR) << "Unexpected ws order list format!" << JsonToString(data);
    return product_order_bundle;
  }

  const auto& info = data["data"].GetObject();
  for (const auto& row : info["rows"].GetArray()) {
    if (row.HasMember("orderStatus")) {
      auto* order = product_order_bundle.add_each_order();
      order->set_market_type(MarketType::Spot);
      order->set_exchange_type(ExchangeType::Prex);
      order->set_api_version("v1");
      std::string exchange_id = row["orderId"].GetString();
      if (row.HasMember("symbol")) {
        order->set_symbol(row["symbol"].GetString());
      } else {
        auto* oc = GetOcm()->FindByExchangeOrderId(exchange_id);
        if (oc != nullptr) {
          order->set_symbol(oc->order_spec().product);
          DLOG(INFO) << "no symbol, set oc symbol:" << oc->order_spec().product;
        }
      }
      if (row.HasMember("price")) {
        order->set_price(stod(row["price"].GetString()));
      }
      if (row.HasMember("orderQty")) {
        order->set_qty(stod(row["orderQty"].GetString()));
      }
      if (row.HasMember("side")) {
        order->set_side(GetOrderSide(row["side"].GetString()));
      }
      if (row.HasMember("orderType")) {
        order->set_order_type(GetOrderType(row["orderType"].GetString()));
      }
      order->set_exchange_order_id(exchange_id);
      order->set_proc_order_id(GetProcOrderId(row));
      if (row.HasMember("transactTime")) {
        order->set_timestamp(stoll(row["transactTime"].GetString()));
      } else {
        order->set_timestamp(GetCurrentTimestamp());
      }
      SetOrderStatus(row["orderStatus"].GetString(), order);
    }
  }
  return product_order_bundle;
}

ProductFillElement PrexParser::ParseFillInfo(const JsonValue& data) {
  /*
  {
    "hasNext": true,
    "history": [
        {
            "execId": "e27467ab-28b4-6b11-38c5-99f6cebf87fb",
            "execType": "EXEC_TYPE_TRADE",
            "orderId": "1ab96be0-87f4-1c16-1ab9-6be087f41c16",
            "clientOrderId": "asdf12345",
            "accountId": "9800101",
            "symbol": "BTCUSD.PERP",
            "side": "ORDER_SIDE_SELL",
            "orderType": "ORDER_TYPE_LIMIT",
            "price": "9083.500000000000000000",
            "orderQty": "1000.000000000000000000",
            "timeInForce": "TIME_IN_FORCE_GOOD_TILL_CANCEL",
            "execInst": "EXEC_INST_INVALID",
            "specialFlag": "SPECIAL_FLAG_INVALID",
            "text": "Web",
            "settlementCurrency": "",
            "orderTime": 1593417232606428367,
            "transactTime": 1593417232606428367,
            "orderStatus": "ORDER_STATUS_FILLED",
            "leavesQty": "0.000000000000000000",
            "cumQty": "1000.000000000000000000",
            "avgPrice": "9083.5",
            "matchId": "6f72436a-72f4-131f-6d78-27bebc58f84b",
            "matchTime": 1593417233189063471,
            "aggressorIndicator": "",
            "lastPrice": "9083.500000000000000000",
            "lastQty": "192.000200000000000000",
            "execFee": "-174.403381670000000000",
            "feeRate": "-0.000100000000000000"
        }
    ]
  }
  */
  ProductFillElement fill;
  fill.set_symbol(data["symbol"].GetString());
  fill.set_fill_timestamp(stoll(data["transactTime"].GetString()));
  fill.set_exchange_order_id(data["orderId"].GetString());
  fill.set_proc_order_id(GetProcOrderId(data));
  fill.set_price(stod(data["lastPrice"].GetString()));
  fill.set_qty(stod(data["lastQty"].GetString()));
  fill.set_fill_id(data["execId"].GetString());
  fill.set_fee(stod(data["execFee"].GetString()));
  if (data["aggressorIndicator"].IsString()) {
    fill.set_fill_type(FillType::UNKNOWN_FILL_TYPE);
  } else if (data["aggressorIndicator"].IsBool()) {
    bool fill_type = data["aggressorIndicator"].GetBool();
    if (fill_type) {
      fill.set_fill_type(FillType::TAKER_FILL_TYPE);
    } else {
      fill.set_fill_type(FillType::MAKER_FILL_TYPE);
    }
  }
  return fill;
}

void PrexParser::ParseRestFillOrders() {
  JsonDocument data;
  if (!PrepareJsonDocument(&data)) {
    LOG(ERROR) << "[ParseRestFillOrders] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  if (data.IsObject() && IsPrexErrMessage(data)) {
    LOG(ERROR) << "[ParseRestFillOrders] " << JsonToString(data);
    return;
  }

  if (!data.HasMember("history") || !data["history"].IsArray()) {
    LOG(ERROR) << "[ParseRestFillOrders] Unexpected fill list format!" << JsonToString(data);
    return;
  }

  auto product_fill_bundle = result_.mutable_fill_list();
  product_fill_bundle->Clear();
  for (const auto& info : data["history"].GetArray()) {
    if (std::string_view(info["execType"].GetString()) == "EXEC_TYPE_TRADE") {
      auto* fill = product_fill_bundle->add_each_fill();
      auto new_fill = ParseFillInfo(info);
      fill->MergeFrom(new_fill);
    }
  }
  if (result_.fill_list().each_fill().size() > 0) {
    result_.set_type(MsgType::ORDER_FILL);
  }
}

ProductFillBundle PrexParser::ParseWsFillInfoListV2(const JsonValue& data) {
  /*
  {
    "topic":"private/execution",
    "data":{
        "actionType":"INSERT",
        "rows":[
            {
                "execId":"ce579a14-04fa-1924-a356-50a7e35c84d8",
                "execType":"EXEC_TYPE_NEW",
                "orderId":"064a46c8-41ed-1c16-064a-46c841ed1c16",
                "clientOrderId":"asdf1234",
                "accountId":"9800101",
                "symbol":"BTCUSD.PERP",
                "side":"ORDER_SIDE_BUY",
                "orderType":"ORDER_TYPE_LIMIT",
                "price":"9113",
                "orderQty":"100",
                "timeInForce":"TIME_IN_FORCE_GOOD_TILL_CANCEL",
                "execInst":"EXEC_INST_PARTICIPATE_DO_NOT_INITIATE",
                "text":"Web",
                "settlementCurrency":"USD",
                "orderTime":"1593409234973194802",
                "transactTime":"1593409234973194802",
                "orderStatus":"ORDER_STATUS_NEW",
                "leavesQty":"100",
                "cumQty":"0",
                "avgPrice":"0",
                "lastPrice":"0",
                "lastQty":"0",
                "execFee":"0",
                "feeRate":"0",
                "specialFlag":"SPECIAL_FLAG_INVALID",
                "matchId":"",
                "matchTime":"0",
                "aggressorIndicator":false
            }]
    }
  }
  */
  ProductFillBundle product_fill_bundle;
  if (data.IsObject() && IsPrexErrMessage(data)) {
    LOG(ERROR) << "[ParseWsFillInfoList] " << JsonToString(data);
    return product_fill_bundle;
  }

  if (!data.HasMember("data") || !data["data"].HasMember("rows") ||
      !data["data"]["rows"].IsArray()) {
    LOG(ERROR) << "Unexpected ws fill list format!" << JsonToString(data);
    return product_fill_bundle;
  }

  for (const auto& row : data["data"]["rows"].GetArray()) {
    if (std::string_view(row["execType"].GetString()) == "EXEC_TYPE_TRADE") {
      auto* fill = product_fill_bundle.add_each_fill();
      auto new_fill = ParseFillInfo(row);
      fill->MergeFrom(new_fill);
    }
  }
  return product_fill_bundle;
}

nlohmann::json PrexParser::ParseHeader(const std::shared_ptr<AsioHttpContext>& http_context) {
  const auto& header = http_context->res.base();
  nlohmann::json json;
  if (header.count("X-Ratelimit-Limit") > 0) {
    json["X-Ratelimit-Limit"] = std::stoi(header.at("X-Ratelimit-Limit").data());
  }
  if (header.count("X-Ratelimit-Remaining") > 0) {
    json["X-Ratelimit-Remaining"] = std::stoi(header.at("X-Ratelimit-Remaining").data());
  }
  if (header.count("X-Ratelimit-Reset") > 0) {
    json["X-Ratelimit-Reset"] = std::stoi(header.at("X-Ratelimit-Reset").data());
  }
  return json;
}

void PrexParser::ParseSubmitResponse() {
  const auto order_context = msg_->oc;
  const auto& http_context = msg_->http_context;
  if (!IsValidResponse(http_context)) {
    return;
  }

  result_.set_type(MsgType::SUBMIT_RESPONSE);
  auto resp = result_.mutable_submit_response();
  resp->Clear();

  resp->set_symbol(order_context->order_spec().product);
  resp->set_proc_order_id(order_context->proc_order_id());
  resp->set_http_status(http_context->res.result_int());
  resp->set_timestamp(GetCurrentTimestamp());

  if (http_context->status == AsioHttpContextStatus::kError) {
    LOG(ERROR) << "[PrexSubmitOrder] AsioHttpContextStatus Error";
    // result: 420 - CUSTOMIZED: NO AVAILABLE CONNECTION
    auto error_code = http_context->res.result_int() == 420
                          ? coin::proto::OrderErrorCode::ORDER_API_ERROR
                          : coin::proto::OrderErrorCode::ORDER_UNCERTAIN;
    auto error_msg = http_context->res.result_int() == 420
                         ? "Json parse error: NO_AVAILABLE_CONNECTION"
                         : "Json parse error: ORDER_NETWORK_ERROR";
    resp->set_error_code(error_code);
    resp->set_exchange_error_msg(error_msg);
    resp->set_success(false);
    return;
  }

  if (http_context->res.result_int() == 429) {
    LOG(ERROR) << "[PrexSubmitOrder] API_LIMIT_ERROR";
    resp->set_error_code(coin::proto::OrderErrorCode::API_LIMIT_ERROR);
    auto rate_info = ParseHeader(http_context);
    resp->set_exchange_error_msg(rate_info.dump());
    resp->set_success(false);
    return;
  }

  rapidjson::Document json;
  json.Parse<rapidjson::ParseFlag::kParseNumbersAsStringsFlag>(http_context->res.body().data());
  if (json.HasParseError()) {
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_API_ERROR);
    resp->set_success(false);
    LOG(ERROR) << "[PrexSubmitOrder] Parse Error\n" << http_context->res;
    return;
  }

  if (json.HasMember("error")) {
    auto error_code = GetOrderErrorCodeFromPrex(json["error"]);
    resp->set_error_code(error_code);
    resp->set_exchange_error_msg(GetPrexExchangeOrderErrorMsg(json));
    if (error_code != coin::proto::OrderErrorCode::ORDER_NO_ERROR) {
      resp->set_success(false);
      LOG(ERROR) << "[PrexSubmitOrder] Error Response" << http_context->res;
    }
    resp->set_success(false);
    return;
  }

  resp->set_success(true);
  resp->set_exchange_order_id(json["orderId"].GetString());

  // handle cancelled order due to timeInForce, like IOC/FOK, PostOnly
  if (std::string_view(json["orderStatus"].GetString()) == "ORDER_STATUS_CANCELED") {
    auto pob = result_.mutable_order_list();
    pob->Clear();
    auto product_order_element = pob->add_each_order();
    product_order_element->set_market_type(MarketType::Spot);
    product_order_element->set_exchange_type(ExchangeType::Prex);
    product_order_element->set_api_version("v1");
    product_order_element->set_exchange_order_id(json["orderId"].GetString());
    product_order_element->set_symbol(order_context->order_spec().product);
    product_order_element->set_price(order_context->order_spec().price);
    product_order_element->set_qty(order_context->order_spec().qty);
    product_order_element->set_proc_order_id(order_context->proc_order_id());
    product_order_element->set_side(order_context->order_spec().order_side);
    product_order_element->set_order_type(order_context->order_spec().order_type);
    product_order_element->set_timestamp(msg_->timestamp);
    product_order_element->set_cancel_confirmed(true);
    result_.set_type(MsgType::ORDER_STATUS);
  }
}

void PrexParser::ParseCancelResponse() {
  /*
  EMPTY
  */
  const auto order_context = msg_->oc;
  const auto& http_context = msg_->http_context;
  if (!IsValidResponse(http_context)) {
    return;
  }

  result_.set_type(MsgType::CANCEL_RESPONSE);
  auto resp = result_.mutable_cancel_response();
  resp->Clear();

  resp->set_symbol(order_context->order_spec().product);
  resp->set_proc_order_id(order_context->proc_order_id());
  resp->set_http_status(http_context->res.result_int());
  resp->set_exchange_order_id(order_context->exchange_order_id());
  resp->set_timestamp(GetCurrentTimestamp());

  if (http_context->status == AsioHttpContextStatus::kError) {
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_NETWORK_ERROR);
    resp->set_success(false);
    LOG(ERROR) << "[PrexCancelOrder] AsioHttpContextStatus Error";
    return;
  }

  if (http_context->res.result_int() == 429) {
    LOG(ERROR) << "[PrexCancelOrder] API_LIMIT_ERROR";
    resp->set_error_code(coin::proto::OrderErrorCode::API_LIMIT_ERROR);
    auto rate_info = ParseHeader(http_context);
    resp->set_exchange_error_msg(rate_info.dump());
    resp->set_success(false);
    return;
  }

  rapidjson::Document json;
  json.Parse<rapidjson::ParseFlag::kParseNumbersAsStringsFlag>(http_context->res.body().data());
  if (json.HasParseError()) {
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_API_ERROR);
    resp->set_success(false);
    LOG(ERROR) << "[PrexCancelOrder] Parse Error\n" << http_context->res;
    return;
  }

  if (json.HasMember("error")) {
    auto error_code = GetOrderErrorCodeFromPrex(json["error"]);
    resp->set_error_code(error_code);
    resp->set_exchange_error_msg(GetPrexExchangeOrderErrorMsg(json));
    if (error_code != coin::proto::OrderErrorCode::ORDER_NO_ERROR) {
      resp->set_success(false);
      LOG(ERROR) << "[PrexCancelOrder] Error Response" << http_context->res;
    }
    return;
  }

  resp->set_success(true);
  if (std::string_view(json["orderStatus"].GetString()) == "ORDER_STATUS_CANCELED") {
    resp->set_cancel_confirmed(true);
  }
  return;
}

void PrexParser::ParseMultiCancelResponse() {
  /*
  [
    {
      "cancelOrders": [
        {
          "symbol": "BTCUSD.PERP",
          "orderId": "b0e189da-2b9a-2637-ec01-c6a62347d662",
          "clientOrderId": "1620820006750442050",
          "side": "ORDER_SIDE_BUY",
          "orderType": "ORDER_TYPE_LIMIT",
          "price": "56512.9",
          "orderQty": "0.001",
          "timeInForce": "TIME_IN_FORCE_GOOD_TILL_CANCEL",
          "execInst": "EXEC_INST_PARTICIPATE_DO_NOT_INITIATE",
          "orderTime": 1620820053104090400,
          "transactTime": 1620820053984745000,
          "orderStatus": "ORDER_STATUS_CANCELED",
          "leavesQty": "0.001",
          "cumQty": "0",
          "avgPrice": "0",
          "cumFee": "0",
          "orderCancelReason": "ORDER_CANCEL_REASON_USER_REQUESTED",
          "specialFlag": "SPECIAL_FLAG_INVALID",
          "text": ""
        },
        {
          "symbol": "BTCUSD.PERP",
          "orderId": "91731f33-bf5f-f8b7-c691-160a82c2f683",
          "clientOrderId": "1620820006750442049",
          "side": "ORDER_SIDE_SELL",
          "orderType": "ORDER_TYPE_LIMIT",
          "price": "56569.3",
          "orderQty": "0.001",
          "timeInForce": "TIME_IN_FORCE_GOOD_TILL_CANCEL",
          "execInst": "EXEC_INST_PARTICIPATE_DO_NOT_INITIATE",
          "orderTime": 1620820053138524000,
          "transactTime": 1620820053984745000,
          "orderStatus": "ORDER_STATUS_CANCELED",
          "leavesQty": "0.001",
          "cumQty": "0",
          "avgPrice": "0",
          "cumFee": "0",
          "orderCancelReason": "ORDER_CANCEL_REASON_USER_REQUESTED",
          "specialFlag": "SPECIAL_FLAG_INVALID",
          "text": ""
        }
      ]
    }
  ]
  */
  const auto& http_context = msg_->http_context;
  if (!IsValidResponse(http_context)) {
    return;
  }
  result_.set_type(MsgType::MULTI_CANCEL_RESPONSE);
  auto multi_cancel_resp = result_.mutable_multi_cancel_response();
  multi_cancel_resp->Clear();

  rapidjson::Document json;
  bool is_success = true;
  auto error_code = coin::proto::OrderErrorCode::ORDER_UNKNOWN_ERROR;
  auto error_msg = std::string{};
  if (http_context->status == AsioHttpContextStatus::kError) {
    LOG(ERROR) << "[PrexMultiCancelOrder] AsioHttpContextStatus Error";
    is_success = false;
    error_code = coin::proto::OrderErrorCode::ORDER_NETWORK_ERROR;
  } else if (http_context->res.result_int() == 429) {
    LOG(ERROR) << "[PrexMultiCancelOrder] API_LIMIT_ERROR";
    auto rate_info = ParseHeader(http_context);
    is_success = false;
    error_code = coin::proto::OrderErrorCode::API_LIMIT_ERROR;
    error_msg = rate_info.dump();
  } else {
    json.Parse<rapidjson::ParseFlag::kParseNumbersAsStringsFlag>(http_context->res.body().data());
    if (json.HasParseError()) {
      LOG(ERROR) << "[PrexMultiCancelOrder] Parse Error\n" << http_context->res;
      is_success = false;
      error_code = coin::proto::OrderErrorCode::ORDER_API_ERROR;
    } else if (json.HasMember("error")) {
      error_code = GetOrderErrorCodeFromPrex(json["error"]);
      error_msg = GetPrexExchangeOrderErrorMsg(json);
      if (error_code != coin::proto::OrderErrorCode::ORDER_NO_ERROR) {
        LOG(ERROR) << "[PrexMultiCancelOrder] Error Response" << http_context->res;
        is_success = false;
      }
    }
  }
  for (const auto& order_context : msg_->oc_list) {
    auto resp = multi_cancel_resp->add_each_cancel_response();
    resp->set_symbol(order_context->order_spec().product);
    resp->set_proc_order_id(order_context->proc_order_id());
    resp->set_http_status(http_context->res.result_int());
    resp->set_exchange_order_id(order_context->exchange_order_id());
    resp->set_timestamp(GetCurrentTimestamp());
    if (!is_success) {
      resp->set_error_code(error_code);
      resp->set_exchange_error_msg(error_msg);
      resp->set_success(false);
      continue;
    }
    resp->set_success(false);
    for (auto& cancel_order : json["cancelOrders"].GetArray()) {
      auto order_id = cancel_order["orderId"].GetString();
      if (order_id != resp->exchange_order_id()) {
        continue;
      }
      resp->set_success(true);
      if (std::string_view(cancel_order["orderStatus"].GetString()) == "ORDER_STATUS_CANCELED") {
        resp->set_cancel_confirmed(true);
      }
      break;
    }
    if (!resp->success()) {
      resp->set_error_code(coin::proto::OrderErrorCode::ORDER_UNKNOWN_ERROR);
    }
  }
}

void PrexParser::ParseAmendResponse() {
  const auto order_context = msg_->oc;
  const auto& http_context = msg_->http_context;
  if (!IsValidResponse(http_context)) {
    return;
  }

  result_.set_type(MsgType::SUBMIT_RESPONSE);
  auto submit_resp = result_.mutable_submit_response();
  submit_resp->Clear();
  submit_resp->set_symbol(order_context->order_spec().product);
  submit_resp->set_proc_order_id(order_context->proc_order_id());
  submit_resp->set_http_status(http_context->res.result_int());
  submit_resp->set_timestamp(msg_->timestamp);

  if (http_context->status == AsioHttpContextStatus::kError) {
    LOG(ERROR) << "[PrexAmendOrder] AsioHttpContextStatus Error";
    submit_resp->set_error_code(coin::proto::OrderErrorCode::ORDER_NETWORK_ERROR);
    submit_resp->set_success(false);
    return;
  }
  if (http_context->res.result_int() == 429) {
    LOG(ERROR) << "[PrexAmendOrder] API_LIMIT_ERROR";
    submit_resp->set_error_code(coin::proto::OrderErrorCode::API_LIMIT_ERROR);
    auto rate_info = ParseHeader(http_context);
    submit_resp->set_exchange_error_msg(rate_info.dump());
    submit_resp->set_success(false);
    return;
  }

  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[PrexAmendOrder] Parsing Error." << msg_->DebugString();
    submit_resp->set_error_code(coin::proto::OrderErrorCode::ORDER_API_ERROR);
    submit_resp->set_success(false);
    return;
  }

  if (document.HasMember("error")) {
    auto error_code = GetOrderErrorCodeFromPrex(document["error"]);
    submit_resp->set_error_code(error_code);
    submit_resp->set_exchange_error_msg(GetPrexExchangeOrderErrorMsg(document));
    if (error_code != coin::proto::OrderErrorCode::ORDER_NO_ERROR) {
      submit_resp->set_success(false);
      LOG(ERROR) << "[PrexAmendOrder] Error Response" << http_context->res;
    }
    return;
  }

  submit_resp->set_success(true);
  submit_resp->set_exchange_order_id(document["orderId"].GetString());
}

void PrexParser::ParseConversionResponse() {
  result_.set_type(MsgType::IGNORABLE);  // default ignore fail conversion response
  const auto& http_context = msg_->http_context;
  if (!IsValidResponse(http_context)) {
    return;
  }

  if (http_context->status == AsioHttpContextStatus::kError) {
    LOG(ERROR) << "[ParseConversionResponse] AsioHttpContextStatus Error";
    return;
  }

  if (http_context->res.result_int() == 429) {
    LOG(ERROR) << "[ParseConversionResponse] API_LIMIT_ERROR";
    return;
  }

  JsonDocument json;
  if (!PrepareJsonDocument(&json)) {
    LOG(ERROR) << "[ParseConversionResponse] Parse Error\n" << http_context->res;
    return;
  }

  if (json.HasMember("error")) {
    LOG(ERROR) << "[ParseConversionResponse] Error Response\n" << http_context->res;
    return;
  }
  if (!json.HasMember("currencyConversionId") || !json.HasMember("conversionStatus")) {
    return;
  }
  std::string conversion_status = json["conversionStatus"].GetString();
  if (conversion_status != "CURRENCY_CONVERSION_STATUS_COMPLETED") {
    // ignore incompleted conversion
    return;
  }
  std::string conversion_id = json["currencyConversionId"].GetString();
  if (conversion_ids_cache_.count(conversion_id)) {
    // this conversion was handled already
    return;
  }
  conversion_ids_cache_.emplace(conversion_id);

  result_.set_type(MsgType::CONVERSION);
  auto* conversion_bundle = result_.mutable_conversion_list();
  conversion_bundle->Clear();
  conversion_bundle->set_market_type(MarketType::Spot);
  conversion_bundle->set_exchange_type(ExchangeType::Prex);
  auto* conversion = conversion_bundle->add_each_conversion();
  conversion->MergeFrom(ParseConversion(json));
}

ConversionElement PrexParser::ParseConversion(const JsonValue& data) {
  ConversionElement conversion;
  conversion.set_conversion_id(data["currencyConversionId"].GetString());
  conversion.set_from_currency(data["fromCurrency"].GetString());
  conversion.set_to_currency(data["toCurrency"].GetString());
  conversion.set_from_amount(stod(data["fromAmount"].GetString()));
  conversion.set_to_amount(stod(data["toAmount"].GetString()));
  conversion.set_exchange_rate(stod(data["exchangeRate"].GetString()));
  conversion.set_timestamp(msg_->timestamp);
  return conversion;
}

void PrexParser::ParseRestHistConversions() {
  JsonDocument data;
  if (!PrepareJsonDocument(&data)) {
    LOG(ERROR) << "[ParseRestHistConversions] Parsing Error\n" << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  if (data.IsObject() && IsPrexErrMessage(data)) {
    LOG(ERROR) << "[ParseRestHistConversions] Error Response\n" << JsonToString(data);
    return;
  }
  if (!data.HasMember("history") || !data["history"].IsArray()) {
    LOG(ERROR) << "[ParseRestHistConversions] Unexpected conversion response format\n"
               << JsonToString(data);
    return;
  }

  result_.set_type(MsgType::CONVERSION);
  auto* conversion_bundle = result_.mutable_conversion_list();
  conversion_bundle->Clear();
  conversion_bundle->set_market_type(MarketType::Spot);
  conversion_bundle->set_exchange_type(ExchangeType::Prex);
  for (const auto& info : data["history"].GetArray()) {
    if (!info.HasMember("currencyConversionId") || !info.HasMember("conversionStatus")) {
      continue;
    }
    std::string conversion_status = info["conversionStatus"].GetString();
    if (conversion_status != "CURRENCY_CONVERSION_STATUS_COMPLETED") {
      // ignore incompleted conversion
      continue;
    }
    std::string conversion_id = info["currencyConversionId"].GetString();
    if (conversion_ids_cache_.count(conversion_id)) {
      // this conversion was handled already
      continue;
    }
    conversion_ids_cache_.emplace(conversion_id);
    auto* conversion = conversion_bundle->add_each_conversion();
    conversion->MergeFrom(ParseConversion(info));
  }
  if (is_first_query_hist_conversions_) {
    is_first_query_hist_conversions_ = false;
    conversion_bundle->Clear();
    result_.set_type(MsgType::IGNORABLE);
  }
}

void PrexParser::ParseWsAccountBalance(const JsonValue& document) {
  auto acc_balance = ParseWsBalance(document);
  if (!is_account_balance_initialized_) {
    // just igore if balance is not initialized
    result_.set_type(MsgType::IGNORABLE);
  }
  if (!acc_balance.has_exchange() || !acc_balance.has_market_type()) {
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  result_.set_type(MsgType::ACCOUNT_BALANCE);
  *result_.mutable_account_balance() = std::move(acc_balance);
}

void PrexParser::ParseWsMessage() {
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseWsMessage] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  if (IsWallet(document)) {
    ParseWsAccountBalance(document);
  } else if (IsOrderUpdate(document)) {
    *result_.mutable_order_list() = ParseWsOrderInfoListV2(document);
    if (result_.order_list().each_order().size() > 0) {
      result_.set_type(MsgType::ORDER_STATUS);
    }
  } else if (IsExecutionUpdate(document)) {
    *result_.mutable_order_list() = ParseWsOrderInfoListV3(document);
    if (result_.order_list().each_order().size() > 0) {
      result_.set_type(MsgType::ORDER_STATUS);
    }
    *result_.mutable_fill_list() = ParseWsFillInfoListV2(document);
    if (result_.fill_list().each_fill().size() > 0) {
      result_.set_type(MsgType::ORDER_FILL);
    }
  } else {
    LOG(WARNING) << "Unknown ws_msg_type: " << JsonToString(document);
  }
}

auto PrexParser::ParseRawMessage(const RawMessage& msg) -> const ParseResult& {
  result_.Clear();
  msg_ = &msg;
  sub_topic_id_ = static_cast<OrderSubTopicId>(msg.type);

  if (msg.type == OrderSubTopicId::WS_MESSAGE) {
    ParseWsMessage();
    return result_;
  }

  switch (msg.http_context->query_type) {
    case OrderSubTopicId::SUBMIT_RESPONSE:
      ParseSubmitResponse();
      break;
    case OrderSubTopicId::CANCEL_RESPONSE:
      ParseCancelResponse();
      break;
    case OrderSubTopicId::AMEND_RESPONSE:
      ParseAmendResponse();
      break;
    case OrderSubTopicId::MULTI_CANCEL_RESPONSE:
      ParseMultiCancelResponse();
      break;
    case OrderSubTopicId::CONVERSION_RESPONSE:
      ParseConversionResponse();
      break;
    case OrderSubTopicId::REST_OPEN_ORDERS:
      ParseRestOpenOrders();
      break;
    case OrderSubTopicId::REST_ALL_ORDERS:
      ParseRestAllOrders();
      break;
    case OrderSubTopicId::REST_FILL_ORDERS:
      ParseRestFillOrders();
      break;
    case OrderSubTopicId::REST_ACCOUNT_BALANCE:
      ParseRestAccountBalance();
      break;
    case OrderSubTopicId::REST_HIST_CONVERSIONS:
      ParseRestHistConversions();
      break;
    default:
      LOG(ERROR) << "Parsing Error. " << msg_->DebugString();
      break;
  }

  return result_;
}

}  // namespace coin2::exchange::prex::order_v1
